package mqtt

import (
	"encoding/json"
	"errors"
	"fmt"
	"iot-base/common/common"
	"iot-base/common/config"
	"iot-base/common/logger"
	"strings"
	"sync"
	"time"

	MQTT "github.com/eclipse/paho.mqtt.golang"
)

type RequestCallBack func(client IOTMQTTBroker, topic string, pCmd *Cmd)
type ReplyCallBackTimeout func(client IOTMQTTBroker, pCmd *Cmd)
type ReplyCallBack func(client IOTMQTTBroker, pCmdResponse *CmdResponse, pCmd *Cmd)

//IOTMQTTBroker mqtt连接
type IOTMQTTBroker interface {
	//Connect(addr string, username string, pwd string, reconnect bool)
	RegisterTopic(topic string, callBack RequestCallBack, template interface{})
	SendCmd(pCmd *Cmd, topic string, timeout int, body interface{}, fun ReplyCallBack, funtimeout ReplyCallBackTimeout)
	SendCmdWait(pCmd *Cmd, topic string, timeout int) (error, bool, interface{})
	Reply(rtopic string, pCmd *Cmd, code int, errString string, body interface{})
	Reconnect()
}

type topicInfo struct {
	topic        string
	fun          RequestCallBack
	qos          byte
	bodyTemplate interface{}
}

type requestInfo struct {
	topic        string
	pCmd         *Cmd
	timeout      int64
	funReply     ReplyCallBack
	funTimeout   ReplyCallBackTimeout
	bodyTemplate interface{}
	qos          byte
}
type CLIENT_TYPE MQTT.Client

//IOTMQTTClient mqttclient 结构
type IOTMQTTClient struct {
	topicsChan chan *topicInfo
	cmdChan    chan *requestInfo
	replyChan  chan *CmdResponse
	topics     []*topicInfo
	mapRequest map[int64]*requestInfo
	reconnect  bool
	//client     MQTT.Client
	addr string
	id   string
	//CLIENT_TYPE
	client    MQTT.Client
	firstConn bool
	un        string
	pwd       string
	pOpts     *MQTT.ClientOptions
}

//RegisterTopic 注册主题
func (self *IOTMQTTClient) RegisterTopic(topic string, callBack RequestCallBack, body interface{}) {
	topicInfo := &topicInfo{
		topic:        topic,
		fun:          callBack,
		qos:          1,
		bodyTemplate: body,
	}
	self.topicsChan <- topicInfo
}

//SendCmd 发送命令
func (self *IOTMQTTClient) SendCmd(pCmd *Cmd, topic string, timeout int, bodyReply interface{}, fun ReplyCallBack, funtimeout ReplyCallBackTimeout) {
	requestInfo := &requestInfo{
		topic:        topic,
		pCmd:         pCmd,
		funReply:     fun,
		funTimeout:   funtimeout,
		bodyTemplate: bodyReply,
		qos:          1,
	}
	//uuid, _ := uuid.NewV4()
	//suuid := fmt.Sprintf("%v", uuid)

	pCmd.SourceID = self.id
	requestInfo.pCmd.CmdID, _ = common.GetSnowflakeIdDefualtID()

	if fun == nil && funtimeout == nil {

		requestInfo.pCmd.TimeStamp = time.Now().Unix()
		//self.mapRequest[requestInfo.pCmd.CmdID] = requestInfo
		body, _ := json.Marshal(requestInfo.pCmd)
		self.publish(requestInfo.topic, 1, false, body)

		//fmt.Println("send msg--------------", requestInfo.topic)
		return
	}
	requestInfo.timeout = int64(timeout)
	self.cmdChan <- requestInfo
	//fmt.Println("---------begin send-----------------------", topic)
}

func (self *IOTMQTTClient) publish(rtopic string, qos byte, isRetain bool, pro []byte) error {
	token := self.client.Publish(rtopic, qos, isRetain, pro)
	token.WaitTimeout(time.Second * 2)
	err := token.Error()
	if err != nil {
		logger.SLog.Error("mqtt token error ", err)
	}
	return err
}

//Reply 回应
func (self *IOTMQTTClient) Reply(rtopic string, pCmd *Cmd, code int, errString string, body interface{}) {
	reply := &CmdResponse{
		CmdType:  pCmd.CmdType + "_res",
		CmdID:    pCmd.CmdID,
		SourceID: self.id,
		ProError: ProError{
			Code:  code,
			Error: errString,
		},
		Body:      body,
		TimeStamp: time.Now().Unix(),
	}
	jreply, err := json.Marshal(reply)
	if err == nil {
		topic := rtopic + "reply/" + pCmd.SourceID
		err := self.publish(topic, 1, false, jreply)
		//token.Wait()
		//if err != nil {
		//fmt.Println("-------------", topic, "err-------2", err)
		//}
		//
		logger.SLog.Info("reply------id-------", err, pCmd.CmdID)
	}

}

func (self *IOTMQTTClient) subTopic(info *topicInfo) {

	if !self.client.IsConnected() {
		return
	}
	//fmt.Println("now subtopic ---------", info.topic)
	token := self.client.Subscribe(info.topic, info.qos, func(client MQTT.Client, msg MQTT.Message) {
		var cmd Cmd
		if info.bodyTemplate != nil {
			cmd.Body, _ = common.CreateInstance(info.bodyTemplate)
		}
		err := json.Unmarshal(msg.Payload(), &cmd)
		//fmt.Println("------------recv------msg---")
		if err == nil {
			//fmt.Println("-------------------topic run-----------------")
			info.fun(self, msg.Topic(), &cmd) //不要阻塞消息队列的线程
			//fmt.Println("-------------------topic run----------end")
		} else {
			//fmt.Println("------------error---------", err)
			logger.SLog.Error("subscribe error topic:", info.topic, " error:", err)
		}

	})
	token.WaitTimeout(time.Duration(10 * time.Second))
	if token.Error() != nil {
		logger.SLog.Error("--------sub----error---------", info.topic, "%%%%%%%%%%%%%", token.Error())
	}
	fmt.Println("--------sub----finished---------", info.topic)

}

func (self *IOTMQTTClient) run() {
	timer := time.NewTimer(10 * time.Second)
	self.mapRequest = make(map[int64]*requestInfo)
	self.topics = make([]*topicInfo, 0, 5)
	for {
		timer.Reset(10 * time.Second)
		select {
		case topicInfo := <-self.topicsChan:
			//fmt.Println("topicinfo handler------------------")
			if topicInfo == nil {
				//fmt.Println("topicinfo handler-------end-----------")
				break
			}
			if !self.client.IsConnected() {
				fmt.Println("topicinfo handler-------end--------2---")
				if topicInfo.topic != "ALL" {
					self.topics = append(self.topics, topicInfo)
				}

				continue
			}
			if topicInfo.topic == "ALL" {
				fmt.Println("-----regist all-------------", len(self.topics))
				if self != nil {
					for _, info := range self.topics {
						fmt.Println("-----regist -------------", info.topic)
						self.subTopic(info)
					}
				}
				fmt.Println("topicinfo handler-------end--------3---")

			} else {
				found := false
				for _, info := range self.topics {
					if info.topic == topicInfo.topic {
						info.fun = topicInfo.fun
						found = true
						break
					}
				}
				if !found {
					self.topics = append(self.topics, topicInfo)
					self.subTopic(topicInfo)
				}
				//fmt.Println("topicinfo handler-------end--------4---")

			}
		case requestInfo := <-self.cmdChan:
			if requestInfo != nil {
				//fmt.Println("************send 1----------", requestInfo.topic)
				requestInfo.pCmd.TimeStamp = time.Now().Unix()
				requestInfo.timeout = requestInfo.timeout + requestInfo.pCmd.TimeStamp
				self.mapRequest[requestInfo.pCmd.CmdID] = requestInfo

				jcmd, _ := json.Marshal(requestInfo.pCmd)
				err := self.publish(requestInfo.topic, 1, false, jcmd)
				logger.SLog.Info("publish ------", requestInfo.topic, " id:", requestInfo.pCmd.CmdID, err)
				if err != nil {
					if strings.Contains(err.Error(), " broken") {
						self.reconnectClient()
					}
				}

				//fmt.Println("************send finish____________", err)
			}
		case pReply := <-self.replyChan:
			if pReply != nil {
				info, ok := self.mapRequest[pReply.CmdID]
				if ok && info != nil {
					if pReply.Body != nil && info.bodyTemplate != nil {
						bodyObj, _ := common.CreateInstance(info.bodyTemplate)
						body, _ := pReply.Body.(*json.RawMessage)
						if body != nil {
							err := json.Unmarshal(*body, bodyObj)
							if err != nil {
								logger.SLog.Error("unmarshal error----type--", pReply.CmdType, pReply.CmdID, err)
							}
							pReply.Body = bodyObj
						}

					}
					info.funReply(self, pReply, info.pCmd)
				} else {
					logger.SLog.Error("get reply but cmdid not found", pReply.CmdID, pReply.CmdType)
				}
				delete(self.mapRequest, pReply.CmdID)

			}
		case <-timer.C:
			now := time.Now().Unix()
			for k, v := range self.mapRequest {
				if v.timeout <= now {
					if v.funTimeout != nil {
						v.funTimeout(self, v.pCmd)
					}
					delete(self.mapRequest, k)

				}

			}

		}
	}
}

//CreateMQTTClient 创建mqtt客户端
func CreateMQTTClient(addr string, un string, pwd string, reconnect bool, id string) IOTMQTTBroker {
	fmt.Println("begin--------------connect to mqtt-------------")
	pClient := &IOTMQTTClient{
		addr:       addr,
		reconnect:  reconnect,
		id:         id,
		topicsChan: make(chan *topicInfo),
		cmdChan:    make(chan *requestInfo),
		replyChan:  make(chan *CmdResponse),
		firstConn:  true,
		un:         un,
		pwd:        pwd,
	}
	opts := MQTT.NewClientOptions().AddBroker(addr)
	pClient.pOpts = opts
	opts.SetAutoReconnect(reconnect)
	opts.SetCleanSession(true)
	opts.SetConnectTimeout(30 * time.Second)
	opts.SetKeepAlive(60 * time.Second)
	//本地测试时需要加上随机数
	//r := rand.New(rand.NewSource(time.Now().UnixNano()))
	//num := r.Intn(128)
	//id += ":" + strconv.Itoa(num)
	opts.SetClientID(id)
	opts.SetUsername(un)
	opts.SetPassword(pwd)
	opts.SetConnectionLostHandler(func(c MQTT.Client, err error) {
		logger.SLog.Error("Connection lost, reason: ", err, " ", addr)
	})
	opts.SetOnConnectHandler(func(client MQTT.Client) {
		logger.SLog.Info("Connection ok, addr: ", addr)
		pClient.client = client
		rtopic := "reply/" + id
		rtopic = config.GetLocalTopic() + rtopic
		logger.SLog.Info("sub topic reply ", rtopic)
		pClient.client.Subscribe(rtopic, 0, func(client MQTT.Client, msg MQTT.Message) {
			body := msg.Payload()
			var reply CmdResponse
			var rbody json.RawMessage
			reply.Body = &rbody
			//var body *json.RawMessage
			err := json.Unmarshal(body, &reply)
			if err == nil {
				logger.SLog.Info("get from-------and sendto chan----: ", rtopic, reply.CmdType, reply.CmdID)
				pClient.replyChan <- &reply
			} else {
				logger.SLog.Error("unmarshal json error", err)
			}

		})
		if !pClient.firstConn {
			logger.SLog.Info("sub topIC all")
			go func() {
				pClient.topicsChan <- &topicInfo{
					topic: "ALL",
				}
				logger.SLog.Info("------sub topIC all--------------")
			}()

		}

		pClient.firstConn = false
		/*fmt.Println("runing register all topic")
		 */
	})

	//使用上面的ClientOptions创建并启动客户端
	pClient.reconnectClient()
	go pClient.run()
	return pClient
}

func (pClient *IOTMQTTClient) reconnectClient() {
	if pClient.client != nil {
		pClient.client.Disconnect(3000)
		logger.SLog.Warn("reconnect mqtt-------------- ")
	}
	pClient.client = MQTT.NewClient(pClient.pOpts)
	if token := pClient.client.Connect(); token.Wait() && token.Error() != nil {
		//fmt.Println("error--------------------", token.Error())
		logger.SLog.Error("mqtt error-------------", token.Error())
	}
}

func (pClient *IOTMQTTClient) Reconnect() {
	pClient.reconnectClient()
}

//SendCmdWait 发送后等待 error错误 bool 是否超时 interface{}返回数据
func (self *IOTMQTTClient) SendCmdWait(pCmd *Cmd, topic string, timeout int) (error, bool, interface{}) {
	var wg sync.WaitGroup
	wg.Add(1)
	var pResponse *CmdResponse
	var istimeout = false
	var err error
	var body interface{}
	self.SendCmd(pCmd, topic, timeout, nil, func(client IOTMQTTBroker, pCmdResponse *CmdResponse, pCmd *Cmd) {
		pResponse = pCmdResponse
		wg.Done()
	}, func(client IOTMQTTBroker, pCmd *Cmd) {
		istimeout = true
		wg.Done()
	})
	wg.Wait()
	if pResponse == nil {
		if istimeout {
			return nil, true, nil
		}
	} else {
		if pResponse.Code == 0 {
			err = nil
			body = pResponse.Body
		} else {
			body = nil
			err = errors.New(pResponse.Error)
		}
	}
	return err, istimeout, body
}
